<?php

namespace Zeke\Net\Http\Middleware;

use Psr\Http\Message\ResponseInterface;
use Zeke\Net\Http\Stream\HttpBodyStream;
use Zeke\Net\Http\Stream\PauseBufferStream;
use Psr\Http\Message\ServerRequestInterface;
use Zeke\Contracts\Io\Stream\ReadableStreamContracts;

/**
 * Limits how many next handlers can be executed concurrently.
 *
 * If this middleware is invoked, it will check if the number of pending
 * handlers is below the allowed limit and then simply invoke the next handler
 * and it will return whatever the next handler returns (or throws).
 *
 * If the number of pending handlers exceeds the allowed limit, the request will
 * be queued (and its streaming body will be paused) and it will return a pending
 * promise.
 * Once a pending handler returns (or throws), it will pick the oldest request
 * from this queue and invokes the next handler (and its streaming body will be
 * resumed).
 *
 * The following example shows how this middleware can be used to ensure no more
 * than 10 handlers will be invoked at once:
 *
 * ```php
 * $server = new StreamingServer(array(
 *     new LimitConcurrentRequestsMiddleware(10),
 *     $handler
 * ));
 * ```
 *
 * Similarly, this middleware is often used in combination with the
 * [`RequestBodyBufferMiddleware`](#requestbodybuffermiddleware) (see below)
 * to limit the total number of requests that can be buffered at once:
 *
 * ```php
 * $server = new StreamingServer(array(
 *     new LimitConcurrentRequestsMiddleware(100), // 100 concurrent buffering handlers
 *     new RequestBodyBufferMiddleware(2 * 1024 * 1024), // 2 MiB per request
 *     new RequestBodyParserMiddleware(),
 *     $handler
 * ));
 * ```
 *
 * More sophisticated examples include limiting the total number of requests
 * that can be buffered at once and then ensure the actual request handler only
 * processes one request after another without any concurrency:
 *
 * ```php
 * $server = new StreamingServer(array(
 *     new LimitConcurrentRequestsMiddleware(100), // 100 concurrent buffering handlers
 *     new RequestBodyBufferMiddleware(2 * 1024 * 1024), // 2 MiB per request
 *     new RequestBodyParserMiddleware(),
 *     new LimitConcurrentRequestsMiddleware(1), // only execute 1 handler (no concurrency)
 *     $handler
 * ));
 * ```
 *
 * 请求处理队列，如果请求并发比较高，则将请求缓冲到连接池中
 * 一个请求处理完之后，从队列中获取请求继续处理
 *
 *
 * @see RequestBodyBufferMiddleware
 */
final class LimitConcurrentRequestsMiddleware
{

    /**
     * 允许的最大值
     * @var int
     */
    private $limit;

    /**
     * 当前请求量
     * @var int
     */
    private $pending = 0;

    /**
     * @var array
     */
    private $queue = array();

    /**
     * @param int $limit Maximum amount of concurrent requests handled.
     *
     * For example when $limit is set to 10, 10 requests will flow to $next
     * while more incoming requests have to wait until one is done.
     */
    public function __construct($limit)
    {
        $this->limit = $limit;
    }


    /**
     *
     *
     * @param ServerRequestInterface $request
     * @param $next
     * @return PromiseInterface|ResponseInterface
     *
     * @throws \Exception
     * @throws \Throwable
     *
     * @author <liu.bin@ruaby.com> 2018-08-31 10:52
     */
    public function __invoke(ServerRequestInterface $request, $next)
    {
        // happy path: simply invoke next request handler if we're below limit
        if ($this->pending < $this->limit) {
            ++$this->pending;

            try {
                $response = $next($request);
            } catch (\Exception $e) {
                $this->processQueue();
                throw $e;
            } catch (\Throwable $e) { // @codeCoverageIgnoreStart
                // handle Errors just like Exceptions (PHP 7+ only)
                $this->processQueue();
                throw $e; // @codeCoverageIgnoreEnd
            }

            // happy path: if next request handler returned immediately,
            // we can simply try to invoke the next queued request
            if ($response instanceof ResponseInterface) {
                //消费掉队列的请求
                $this->processQueue();
                return $response;
            }

            // if the next handler returns a pending promise, we have to
            // await its resolution before invoking next queued request
            // 如果下级返回的仍然是promise,那么需要封装新的promise, 并定义then方法
            $nextPromise = Promise\resolve($response);
            return $this->await($nextPromise);
        }


        // if we reach this point, then this request will need to be queued
        // check if the body is streaming, in which case we need to buffer everything
        // 如果pending >= limit,请求需要排队
        // 如果body仍然有数据，我们需要将数据缓冲下来
        $body = $request->getBody();
        if ($body instanceof ReadableStreamContracts) {
            // pause actual body to stop emitting data until the handler is called
            $size = $body->getSize();
            $body = new PauseBufferStream($body);
            //只接受http数据包，不触发data事件
            $body->pauseImplicit();

            // replace with buffering body to ensure any readable events will be buffered
            $request = $request->withBody(new HttpBodyStream($body, $size));
        }

        // get next queue position
        $queue = &$this->queue;
        $queue[] = null;
        end($queue);

        //定义当前request的队列下标
        $id = key($queue);

        // 定义promise的cancel处理函数
        $deferred = new Deferred(function ($_, $reject) use (&$queue, $id) {
            // queued promise cancelled before its next handler is invoked
            // remove from queue and reject explicitly
            // 在调用下一个处理函数之前，定义好promise的cancel方法处理函数
            unset($queue[$id]);
            $reject(new \RuntimeException('Cancelled queued next handler'));
        });

        // queue request and process queue if pending does not exceed limit
        $queue[$id] = $deferred;
        $pending    = &$this->pending;
        $that       = $this;

        // 创建 promise对象，将Deferred的回调函数注册到promise的cancel调用中
        // 然后定义promise的$resolver回调函数,当deferred->resolver()时，执行then里面的方法
        return $deferred->promise()->then(function () use ($request, $next, $body, &$pending, $that) {
            // invoke next request handler
            ++$pending;

            try {
                //执行下一个中间件，执行BodyBuffer中间件
                $response = $next($request);
            } catch (\Exception $e) {
                $that->processQueue();
                throw $e;
            } catch (\Throwable $e) { // @codeCoverageIgnoreStart
                // handle Errors just like Exceptions (PHP 7+ only)
                $that->processQueue();
                throw $e; // @codeCoverageIgnoreEnd
            }

            // resume readable stream and replay buffered events
            if ($body instanceof PauseBufferStream) {
                //bodyStream继续触发data事件
                $body->resumeImplicit();
            }

            // if the next handler returns a pending promise, we have to
            // await its resolution before invoking next queued request
            // 如果下一级中间件返回的仍然是promise，则说明此此request仍然没有处理结束，
            // 需要注册回调，里面调用processQueue方法，消费掉此次request
            return $that->await(Promise\resolve($response));
        });
    }


    /**
     * 定义下级中间件异步执行resolve时的then回调函数
     * @internal
     * @param PromiseInterface $nextPromise
     * @return PromiseInterface
     */
    public function await(PromiseInterface $nextPromise)
    {
        $that = $this;

        // 此$promise为下级中间件的返回值，此处需要定义下级中间件内执行resolve时的触发函数
        return $nextPromise->then(function ($response) use ($that) {
            //当下级调用promise的then回调后，相当于已处理完一次请求，此处队列减1
            $that->processQueue();

            return $response;
        }, function ($error) use ($that) {
            $that->processQueue();

            return Promise\reject($error);
        });
    }


    /**
     * 消费队列，
     * @internal
     */
    public function processQueue()
    {
        // skip if we're still above concurrency limit or there's no queued request waiting
        if (--$this->pending >= $this->limit || !$this->queue) {
            return;
        }

        // 当前中间件返回的promise
        $first = reset($this->queue);
        // 重置当前队列下标到第一个，并删除
        unset($this->queue[key($this->queue)]);

        // 执行此函数，将会调用promise注册的$resolver函数，也就是then回调
        // resolve处理逻辑：
        // 将队列中，所有请求都执行到BodyBuffer的中间件中(异步)，请求只注册data事件，不执行真正的业务同步代码
        // 将请求直接放入队列后返回，将增加连接的并发量
        $first->resolve();
    }
}
